package main.java.black;

import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import lombok.SneakyThrows;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/**
 * TopIpCountBolt
 *
 * @author zhangyimin
 * @version 1.0
 * @date 2018-11-13 下午3:01
 */
public class BlackListCountMSDBBolt extends BaseRichBolt {

    private OutputCollector collector;

    private Map<String, Long> result = new HashMap<>();

    private static String driver = "com.mysql.jdbc.Driver";

    private static String url = "jdbc:mysql://localhost:3306/hive_etl";


    static {
        try {
            Class.forName(driver);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    @SneakyThrows
    @Override
    public void execute(Tuple tuple) {
        String userId = tuple.getStringByField("user_id");

        Long pv = tuple.getLongByField("pv");

//        int resCount=Db.use().insert(
//                Entity.create("user_ip_result")
//                        .set("user_id", userId)
//                        .set("user_pv", pv)
//        );


//        List<Entity> entities =Db.use().query("insert into user_ip_result(user_id,user_pv) values (?,?) on duplicate key update user_pv = user_pv + ?", userId, pv,pv);
//        Db.use().closeConnection(Db.use().getConnection());

        int resInt = Db.use().insertOrUpdate(Entity.create("user_ip_result").set("user_id", userId)
                .set("user_pv", pv), "user_id");


        if (resInt>0){
            System.out.println("插入成功!");
        }else {
            System.out.println("插入失败!");
        }

        this.collector.emit(new Values(userId, result.get(userId)));

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
